package xyz.hubery.stream;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import xyz.hubery.domain.*;
import xyz.hubery.dto.FreezeDTO;
import xyz.hubery.dto.UpdateAccountDTO;
import xyz.hubery.enums.AsyncTaskStatus;
import xyz.hubery.enums.AsyncTaskType;
import xyz.hubery.enums.OrderStatus;
import xyz.hubery.mapper.AsyncTaskMapper;
import xyz.hubery.mapper.EntrustOrderMapper;
import xyz.hubery.model.Constant;
import xyz.hubery.service.EntrustOrderService;
import xyz.hubery.service.MarketService;
import xyz.hubery.service.TurnoverOrderService;
import xyz.hubery.domain.AsyncTask;

import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

/**
 * 消息队列监听器(消费者)
 */
@Service
@Slf4j
public class MessageConsumerListener {

    @Autowired
    private EntrustOrderService entrustOrderService;
    @Autowired
    private EntrustOrderMapper entrustOrderMapper;
    @Autowired
    private TurnoverOrderService turnoverOrderService;
    @Autowired
    private MarketService marketService;
    @Autowired
    private AsyncTaskMapper asyncTaskMapper;

//    /**
//     * 接收处理完的订单
//     * @return
//     */
//    @StreamListener("completed-orders-in")
//    public void completedOrders(List<Order> completedOrders) {
//        // 修正这些委托单的值
//        entrustOrderService.updatecompletedOrders(completedOrders);
//    }

    /**
     * 接收成交记录
     */
    @StreamListener("exchange-trades-in")
    public void exchangeTrades(List<ExchangeTrade> exchangeTrades) {
        for (ExchangeTrade exchangeTrade : exchangeTrades) {
            try {
                processSingleTrade(exchangeTrade); // 一条记录放在一个事务中处理，当前记录处理失败不会影响到其它记录
            } catch (Exception e) {
                log.error("处理成交记录失败, trade: {}", exchangeTrade, e);
                // TODO 回退当前有效的一条订单给撮合服务还是丢弃》》》
                //继续处理下一条
            }
        }
    }

    /**
     * 处理单条成交记录 - 使用独立事务
     * 1、更新委托单状态和金额
     * 2、生成成交记录
     * 3、更新账户余额
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW) // 每条成交独立事务
    public void processSingleTrade(ExchangeTrade exchangeTrade) {
        log.info("接收到成交记录:{}" + exchangeTrade);
        Long buyOrderId = Long.valueOf(exchangeTrade.getBuyOrderId());
        Long sellOrderId = Long.valueOf(exchangeTrade.getSellOrderId());

        // 1. 检查订单ID是否相同
        if (buyOrderId.equals(sellOrderId)) {
            log.error("买卖订单ID相同，数据异常。buyOrderId: {}, sellOrderId: {}",
                    buyOrderId, sellOrderId);
            throw new IllegalArgumentException("买卖订单ID不能相同");
        }

        // 2、查询委托单数据(行锁)，按固定顺序获取锁避免死锁
        List<Long> orderIds = Arrays.asList(buyOrderId, sellOrderId);
        Collections.sort(orderIds); // 按ID排序避免死锁

        EntrustOrder buyEntrustOrder = null;
        EntrustOrder sellEntrustOrder = null;

        for (Long orderId : orderIds) {
            EntrustOrder order = entrustOrderMapper.selectForUpdate(orderId);
            if (order.getId().equals(buyOrderId)) {
                buyEntrustOrder = order;
            } else {
                sellEntrustOrder = order;
            }
        }

        // 3、检查订单是否存在
        if (buyEntrustOrder == null || sellEntrustOrder == null) {
            log.error("委托单不存在, buyOrderId: {}, sellOrderId: {}", buyOrderId, sellOrderId);
            throw new RuntimeException("委托单不存在");
        }

        // 4、检查订单状态
        if (buyEntrustOrder.getStatus() != OrderStatus.OPEN.getCode() ||
                sellEntrustOrder.getStatus() != OrderStatus.OPEN.getCode()) {
            log.warn("委托单状态异常, buyStatus: {}, sellStatus: {}",
                    buyEntrustOrder.getStatus(), sellEntrustOrder.getStatus());
            throw new RuntimeException("委托单状态异常");
        }

        // 5、更新委托单数据
        boolean[] updateds = updateOrderByExchangeTrade(buyEntrustOrder, sellEntrustOrder, exchangeTrade);

        if (!updateds[0] || !updateds[1]) {
            throw new RuntimeException("更新委托单失败，版本冲突");
        }

        // 6、新增成交记录
        TurnoverOrder turnoverOrder = exchangeTrade2TurnoverOrder(exchangeTrade);
        turnoverOrderService.save(turnoverOrder);

        // 7、创建更新账户余额 异步任务
        UpdateAccountDTO accountDTO = exchangeTrade2UpdateAccountDTO(exchangeTrade);
        createAccountUpdateTask(accountDTO);
    }

    /**
     * 根据成交记录更新委托单数据
     *
     * @param buyEntrustOrder
     * @param sellEntrustOrder
     * @param exchangeTrade
     * @return
     */
    private boolean[] updateOrderByExchangeTrade(EntrustOrder buyEntrustOrder, EntrustOrder sellEntrustOrder, ExchangeTrade exchangeTrade) {
        buyEntrustOrder.setDeal(buyEntrustOrder.getDeal().add(exchangeTrade.getVolume())); // 添加本次成交量
        if (buyEntrustOrder.getDeal().compareTo(buyEntrustOrder.getVolume()) >= 0) {
            buyEntrustOrder.setStatus(OrderStatus.COMPLETED.getCode()); // 设置状态为成交
        }
        boolean buyUpdated = entrustOrderService.updateById(buyEntrustOrder);
        sellEntrustOrder.setDeal(sellEntrustOrder.getDeal().add(sellEntrustOrder.getVolume())); // 添加本次成交量
        if (sellEntrustOrder.getDeal().compareTo(sellEntrustOrder.getVolume()) >= 0) {
            sellEntrustOrder.setStatus(OrderStatus.COMPLETED.getCode()); // 设置状态为成交
        }
        boolean sellUpdated = entrustOrderService.updateById(sellEntrustOrder);
        return new boolean[]{buyUpdated, sellUpdated};
    }

    /**
     * 更新委托单数据
     *
     * @param order 委托单
     * @param trade 成交记录
     * @param isBuy 是否买入
     * @return
     */
    private boolean updateOrderWithVersion(EntrustOrder order, ExchangeTrade trade, boolean isBuy) {
        // 计算新的已成交数量
        BigDecimal newFilledVolume = order.getDeal().add(trade.getVolume());

        // 判断订单是否完全成交
        Integer newStatus = newFilledVolume.compareTo(order.getVolume()) >= 0 ?
                OrderStatus.COMPLETED.getCode() : OrderStatus.OPEN.getCode();

        // 乐观锁更新
        int updatedRows = entrustOrderMapper.updateOrderWithVersion(
                order.getId(),
                newFilledVolume,
                newStatus,
                order.getVersion()  // 传入当前版本号
        );

        return updatedRows > 0;
    }


    /**
     * 创建异步任务
     *
     * @param accountDTO
     */
    private void createAccountUpdateTask(UpdateAccountDTO accountDTO) {
        try {
            AsyncTask asyncTask = new AsyncTask(AsyncTaskType.UPDATE_ACCOUNT.getType(), JSON.toJSONString(accountDTO), Constant.TASK_MAX_RETRY);
            asyncTaskMapper.insert(asyncTask);
        } catch (Exception e) {
            log.error("创建资金更新任务失败", e);
            throw new RuntimeException("创建异步任务失败", e);
        }
    }

    /**
     * 类型转换
     *
     * @param exchangeTrade
     * @return
     */
    private UpdateAccountDTO exchangeTrade2UpdateAccountDTO(ExchangeTrade exchangeTrade) {
        if (exchangeTrade == null) {
            return null;
        }

        // 查询市场信息
        Market market = marketService.getMarketBySymbol(exchangeTrade.getSymbol());
        if (market == null) {
            log.error("该交易市场不存在：{}", exchangeTrade.getSymbol());
            return null;
        }

        // 查询委托单信息
        EntrustOrder buyEntrustOrder = entrustOrderService.getById(exchangeTrade.getBuyOrderId()); // 买方委托单
        EntrustOrder sellEntrustOrder = entrustOrderService.getById(exchangeTrade.getSellOrderId()); //卖方

        UpdateAccountDTO accountDTO = new UpdateAccountDTO();
        accountDTO.setBuyUserId(buyEntrustOrder.getUserId());
        accountDTO.setSellUserId(sellEntrustOrder.getUserId());

        accountDTO.setBuyCoinId(market.getBuyCoinId());
        accountDTO.setSellCoinId(market.getSellCoinId());

        accountDTO.setPrice(exchangeTrade.getPrice());
        accountDTO.setVolume(exchangeTrade.getVolume());
        accountDTO.setTurnover(exchangeTrade.getBuyTurnover());

        accountDTO.setBuyFeeRate(buyEntrustOrder.getFeeRate());
        accountDTO.setSellFeeRate(sellEntrustOrder.getFeeRate());

        // 判断委托单是否完成(实际成交量等于委托数量，说明全部交易完成
        byte buyStatus = (byte) (buyEntrustOrder.getDeal().compareTo(buyEntrustOrder.getVolume()) == 0 ? 2 : 1);
        byte sellStatus = (byte) (sellEntrustOrder.getDeal().compareTo(sellEntrustOrder.getVolume()) == 0 ? 2 : 1);
        accountDTO.setBuyStatus(buyStatus);
        accountDTO.setSellStatus(sellStatus);
        if (buyStatus == 2) {
            accountDTO.setBuyFreeze(buyEntrustOrder.getFreeze());
        }
        if (sellStatus == 2) {
            accountDTO.setSellFreeze(sellEntrustOrder.getFreeze());
        }

        return accountDTO;
    }


    /**
     * 类型转换
     *
     * @return
     */
    private TurnoverOrder exchangeTrade2TurnoverOrder(ExchangeTrade exchangeTrade) {

        if (exchangeTrade == null) {
            return null;
        }
        TurnoverOrder turnoverOrder = new TurnoverOrder();

        // 查询市场信息
        Market market = marketService.getMarketBySymbol(exchangeTrade.getSymbol());
        if (market == null) {
            log.error("该交易市场不存在：{}", exchangeTrade.getSymbol());
            return null;
        }
        // 交易市场基本信息
        turnoverOrder.setMarketId(market.getId());
        turnoverOrder.setMarketName(market.getName());
        turnoverOrder.setMarketType(market.getType());
        turnoverOrder.setTradeType(exchangeTrade.getDirection().getCode());
        turnoverOrder.setSymbol(market.getSymbol());

        // 查询委托单信息
        EntrustOrder buyEntrustOrder = entrustOrderService.getById(exchangeTrade.getBuyOrderId()); // 买方委托单
        EntrustOrder sellEntrustOrder = entrustOrderService.getById(exchangeTrade.getSellOrderId()); //卖方

        // 卖方基本信息
        turnoverOrder.setSellUserId(sellEntrustOrder.getUserId());//sellUserId
        turnoverOrder.setSellCoinId(market.getSellCoinId());//sellCoinId
        turnoverOrder.setSellOrderId(sellEntrustOrder.getId());//sellOrderId
        turnoverOrder.setSellPrice(sellEntrustOrder.getPrice());
        turnoverOrder.setSellFeeRate(sellEntrustOrder.getFeeRate());//sellFeeRate
        turnoverOrder.setSellVolume(sellEntrustOrder.getVolume());//sellVolume
        turnoverOrder.setDealSellFee(sellEntrustOrder.getFee());//dealSellFee
        turnoverOrder.setDealSellFeeRate(sellEntrustOrder.getFeeRate());//dealSellFeeRate
        // 买方信息
        turnoverOrder.setBuyUserId(buyEntrustOrder.getUserId());//buyUserId
        turnoverOrder.setBuyCoinId(market.getBuyCoinId());//buyCoinId
        turnoverOrder.setBuyOrderId(buyEntrustOrder.getId());//buyOrderId
        turnoverOrder.setBuyVolume(buyEntrustOrder.getVolume());//buyVolume
        turnoverOrder.setBuyPrice(buyEntrustOrder.getPrice());//buyPrice
        turnoverOrder.setBuyFeeRate(buyEntrustOrder.getFeeRate());//buyFeeRate
        turnoverOrder.setDealBuyFee(buyEntrustOrder.getFee()); //dealBuyFee
        turnoverOrder.setDealBuyFeeRate(buyEntrustOrder.getFeeRate());//dealBuyFeeRate


        turnoverOrder.setOrderId(buyEntrustOrder.getId()); //orderId
        turnoverOrder.setAmount(exchangeTrade.getBuyTurnover());//amount
        turnoverOrder.setVolume(exchangeTrade.getVolume());//volume
        turnoverOrder.setStatus(1);//status
        turnoverOrder.setPrice(exchangeTrade.getPrice());//Price

        return turnoverOrder;
    }


    /**
     * 撤销订单
     *
     * @param orderId
     */
    @Transactional
    @StreamListener("cancel-order-in")
    public void cancelOrder(String orderId) {
        // 查询订单
        EntrustOrder entrustOrder = entrustOrderService.getOne(new LambdaQueryWrapper<EntrustOrder>().eq(EntrustOrder::getId, orderId).last("FOR UPDATE"));
        if (entrustOrder.getStatus() != OrderStatus.OPEN.getCode()) {
            throw new RuntimeException("订单状态不允许删除");
        }
        // 更新
        entrustOrder.setStatus(OrderStatus.CANCELED.getCode());
        BigDecimal freeze = entrustOrder.getFreeze();
        entrustOrder.setFreeze(BigDecimal.ZERO);
        boolean updateById = entrustOrderService.updateById(entrustOrder);
        if(!updateById) {
            throw new RuntimeException("更新委托单状态失败");
        }
        // 解冻该委托单中未处理的部分
        // 不直接远程调用资金服务，因为分布式下不能保证事务


        // 查询货币id
        Market market = marketService.getMarketBySymbol(entrustOrder.getSymbol());
        if (market == null) {
            throw new IllegalArgumentException("交易市场异常");
        }
        Long coinId = null;
        if (entrustOrder.getType() == 1) {
            coinId = market.getBuyCoinId();
        } else {
            coinId = market.getSellCoinId();
        }
        FreezeDTO dto = new FreezeDTO();
        dto.setUserId(entrustOrder.getUserId());
        dto.setCoinId(coinId);
        dto.setFreeze(freeze);
        dto.setOrderId(entrustOrder.getId());
//        accountServiceFeign.unFreeze(dto);
        // 3. 插入异步任务记录(通过定时任务解冻资金
        AsyncTask task = new AsyncTask(AsyncTaskType.UNFREEZE_BALANCE.getType(), JSON.toJSONString(dto), Constant.TASK_MAX_RETRY);
        asyncTaskMapper.insert(task);
    }

}